package io.shdxhl.simplerpc.code.zk;

import io.shdxhl.simplerpc.client.provider.ServiceProvider;
import io.shdxhl.simplerpc.code.cache.SimpleProviderCache;
import io.shdxhl.simplerpc.code.config.SimpleConfiguration;
import org.I0Itec.zkclient.IZkChildListener;
import org.I0Itec.zkclient.IZkDataListener;
import org.I0Itec.zkclient.ZkClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;

@Component
public class ZkUtil {

    @Autowired
    private ZkClient zkClient;

    @Autowired
    private SimpleProviderCache simpleProviderCache;

    @Autowired
    private SimpleConfiguration simpleConfiguration;

    /**
     * 获取根节点的所有子节点
     * @return
     */
    public List<String> getServerList() {
        String path = simpleConfiguration.getZkRoot();
        return zkClient.getChildren(path);
    }

    public List<ServiceProvider> getServerInfos(String serviceName) {
        String path = simpleConfiguration.getZkRoot() + "/" + serviceName;
        List<String> children = zkClient.getChildren(path);
        List<ServiceProvider> providerList = convertToProviderService(serviceName, children);
        return providerList;
    }

    /**
     * 接口服务转换成指定对象
     * @param serviceName
     * @param children
     * @return
     */
    private List<ServiceProvider> convertToProviderService(String serviceName, List<String> children) {
        if(CollectionUtils.isEmpty(children)){
            return new ArrayList<>(0);
        }
        //将服务器节点信息转为服务存储记录对象
        List<ServiceProvider> providerList = children.stream().map(v->{
            String[] serviceInfo = v.split(":");
            return ServiceProvider.builder()
                    .serviceName(serviceName)
                    .serviceIp(serviceInfo[0])
                    .rpcPort(Integer.parseInt(serviceInfo[1]))
                    .build();
        }).collect(Collectors.toList());
        return providerList;
    }

    /**
     * 创建zk根节点
     */
    public void createZkRootNode() {
        boolean exists = zkClient.exists(simpleConfiguration.getZkRoot());
        if(!exists){
            zkClient.createPersistent(simpleConfiguration.getZkRoot());
        }
    }

    /**
     * 创建ZK根节点下其他节点
     * @param path
     */
    public void createZkPersistentNode(String path) {
        String pathName= simpleConfiguration.getZkRoot() + "/" + path;
        boolean exists = zkClient.exists(pathName);
        if(!exists){
            zkClient.createPersistent(pathName);
        }
    }

    /**
     * 创建临时节点
     * @param path
     */
    public void createZKNode(String path) {
        String pathName= simpleConfiguration.getZkRoot() + "/" + path;
        boolean exists = zkClient.exists(pathName);
        if(!exists){
            zkClient.createEphemeral(pathName);
        }
    }

    /**
     * 服务订阅接口
     * @param serviceName
     */
    public void subscribe(String serviceName) {
        String path = simpleConfiguration.getZkRoot() + "/" + serviceName;
        //添加ZK订功能
         zkClient.subscribeChildChanges(path, new IZkChildListener() {
             @Override
             public void handleChildChange(String s, List<String> list) throws Exception {
                 if (!CollectionUtils.isEmpty(list)) {
                     List<ServiceProvider> providerList = convertToProviderService(serviceName, list);
                     //更新缓存记录
                     simpleProviderCache.update(serviceName, providerList);
                 }
             }
         });
    }

}
